
#pragma once

#include <ttg/services.h>
#include <ttg/pipeline.h>

#include "FileLoader.h"
#include "TempFile.h"
#include "HashFunction.h"

using namespace ttg;
using namespace ttg::pipeline;

class ttgLoader :public Node<void *, Buffer<char> *>
{
	private:
		FileLoader *current;
	protected:
		virtual void process(void *data)
		{
			Buffer<char> *res;
			if ((res = current->loadNext()) != NULL)
			{
				sendNext(res);
				sendThis(NULL);
			}
		}
	public:
		ttgLoader(TempFile &file)
		{ current = new FileLoader(file.getFileName()); }

		~ttgLoader()
		{ SAFE_DELETE(current); }
};

class ttgHasher :public Node<Buffer<char> *, unsigned __int64>
{
	virtual void process(Buffer<char> *data)
	{
		sendNext(hashFunc(data));
		SAFE_DELETE(data);
	}
};

class ttgGatherer :public Node<unsigned __int64, unsigned __int64>
{
	private:
		unsigned __int64 hash_value;
	protected:
		virtual void onBegin()
		{ hash_value = 0; }

		virtual void onEnd()
		{ sendNext(hash_value); }

		virtual void process(unsigned __int64 data)
		{ hash_value += data; }
	public:
		ttgGatherer()
			:Node("Gatherer", true)
		{ /*nothing*/ }
};

//Performs parallel file hashing (via ttg) and returns hash-value + elapsed time.
std::pair<unsigned __int64, double> ttgHashing(TempFile &file)
{
	//Creating pipeline.
	static
	Pipeline<void *, unsigned __int64> pl(wrap(new ttgLoader(file)) +
										  wrap(new ttgHasher()) + 
										  wrap(new ttgGatherer()));
	//Launching.
	double t_start = services::ServiceBase::_Timer().getSeconds();
	unsigned __int64 res = pl.run(NULL)[0];
	double t_end = services::ServiceBase::_Timer().getSeconds();
	//Returning formatted result.
	return std::pair<unsigned __int64, double>(res, t_end - t_start);
}
